package com.flink.hbase.sql2hdfs.model;

import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonFormat;
import org.apache.flink.types.Row;

import java.io.Serializable;
import java.time.LocalDateTime;
import java.util.Objects;

/**
 * 用户数据模型
 * 对应 HBase 表中的用户信息
 */
public class UserData implements Serializable {
    private static final long serialVersionUID = 1L;
    
    private String userId;
    private String userName;
    private Integer age;
    private String gender;
    private String email;
    private String phone;
    private String address;
    private String city;
    private String country;
    private String department;
    private String position;
    private Double salary;
    
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
    private LocalDateTime createTime;
    
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
    private LocalDateTime updateTime;
    
    // 处理后的字段
    private String userLevel;          // UDF计算的用户等级
    private String ageGroup;           // UDF计算的年龄组
    private String salaryLevel;        // UDF计算的薪资等级
    private String processTime;        // 处理时间
    private String partitionDate;      // 分区日期
    
    public UserData() {}
    
    public UserData(String userId, String userName, Integer age, String gender, String email,
                   String phone, String address, String city, String country, String department,
                   String position, Double salary, LocalDateTime createTime, LocalDateTime updateTime) {
        this.userId = userId;
        this.userName = userName;
        this.age = age;
        this.gender = gender;
        this.email = email;
        this.phone = phone;
        this.address = address;
        this.city = city;
        this.country = country;
        this.department = department;
        this.position = position;
        this.salary = salary;
        this.createTime = createTime;
        this.updateTime = updateTime;
    }
    
    /**
     * 从 Flink Row 转换为 UserData
     */
    public static UserData fromRow(Row row) {
        UserData userData = new UserData();
        userData.setUserId((String) row.getField(0));
        userData.setUserName((String) row.getField(1));
        userData.setAge((Integer) row.getField(2));
        userData.setGender((String) row.getField(3));
        userData.setEmail((String) row.getField(4));
        userData.setPhone((String) row.getField(5));
        userData.setAddress((String) row.getField(6));
        userData.setCity((String) row.getField(7));
        userData.setCountry((String) row.getField(8));
        userData.setDepartment((String) row.getField(9));
        userData.setPosition((String) row.getField(10));
        userData.setSalary((Double) row.getField(11));
        userData.setCreateTime((LocalDateTime) row.getField(12));
        userData.setUpdateTime((LocalDateTime) row.getField(13));
        return userData;
    }
    
    /**
     * 转换为 Flink Row
     */
    public Row toRow() {
        Row row = new Row(19); // 包含所有字段
        row.setField(0, userId);
        row.setField(1, userName);
        row.setField(2, age);
        row.setField(3, gender);
        row.setField(4, email);
        row.setField(5, phone);
        row.setField(6, address);
        row.setField(7, city);
        row.setField(8, country);
        row.setField(9, department);
        row.setField(10, position);
        row.setField(11, salary);
        row.setField(12, createTime);
        row.setField(13, updateTime);
        row.setField(14, userLevel);
        row.setField(15, ageGroup);
        row.setField(16, salaryLevel);
        row.setField(17, processTime);
        row.setField(18, partitionDate);
        return row;
    }
    
    /**
     * 获取年龄组
     */
    public String calculateAgeGroup() {
        if (age == null) return "未知";
        if (age < 25) return "青年";
        if (age < 35) return "青壮年";
        if (age < 50) return "中年";
        return "中老年";
    }
    
    /**
     * 获取薪资等级
     */
    public String calculateSalaryLevel() {
        if (salary == null) return "未知";
        if (salary < 5000) return "初级";
        if (salary < 10000) return "中级";
        if (salary < 20000) return "高级";
        return "专家";
    }
    
    /**
     * 获取用户等级（基于年龄、薪资等综合计算）
     */
    public String calculateUserLevel() {
        int score = 0;
        
        // 年龄评分
        if (age != null) {
            if (age >= 25 && age <= 45) score += 2; // 工作黄金年龄
            else if (age > 18 && age < 60) score += 1;
        }
        
        // 薪资评分
        if (salary != null) {
            if (salary >= 20000) score += 3;
            else if (salary >= 10000) score += 2;
            else if (salary >= 5000) score += 1;
        }
        
        // 综合等级
        if (score >= 4) return "VIP";
        if (score >= 2) return "高级用户";
        if (score >= 1) return "普通用户";
        return "初级用户";
    }
    
    // Getter and Setter methods
    public String getUserId() { return userId; }
    public void setUserId(String userId) { this.userId = userId; }
    
    public String getUserName() { return userName; }
    public void setUserName(String userName) { this.userName = userName; }
    
    public Integer getAge() { return age; }
    public void setAge(Integer age) { this.age = age; }
    
    public String getGender() { return gender; }
    public void setGender(String gender) { this.gender = gender; }
    
    public String getEmail() { return email; }
    public void setEmail(String email) { this.email = email; }
    
    public String getPhone() { return phone; }
    public void setPhone(String phone) { this.phone = phone; }
    
    public String getAddress() { return address; }
    public void setAddress(String address) { this.address = address; }
    
    public String getCity() { return city; }
    public void setCity(String city) { this.city = city; }
    
    public String getCountry() { return country; }
    public void setCountry(String country) { this.country = country; }
    
    public String getDepartment() { return department; }
    public void setDepartment(String department) { this.department = department; }
    
    public String getPosition() { return position; }
    public void setPosition(String position) { this.position = position; }
    
    public Double getSalary() { return salary; }
    public void setSalary(Double salary) { this.salary = salary; }
    
    public LocalDateTime getCreateTime() { return createTime; }
    public void setCreateTime(LocalDateTime createTime) { this.createTime = createTime; }
    
    public LocalDateTime getUpdateTime() { return updateTime; }
    public void setUpdateTime(LocalDateTime updateTime) { this.updateTime = updateTime; }
    
    public String getUserLevel() { return userLevel; }
    public void setUserLevel(String userLevel) { this.userLevel = userLevel; }
    
    public String getAgeGroup() { return ageGroup; }
    public void setAgeGroup(String ageGroup) { this.ageGroup = ageGroup; }
    
    public String getSalaryLevel() { return salaryLevel; }
    public void setSalaryLevel(String salaryLevel) { this.salaryLevel = salaryLevel; }
    
    public String getProcessTime() { return processTime; }
    public void setProcessTime(String processTime) { this.processTime = processTime; }
    
    public String getPartitionDate() { return partitionDate; }
    public void setPartitionDate(String partitionDate) { this.partitionDate = partitionDate; }
    
    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        UserData userData = (UserData) o;
        return Objects.equals(userId, userData.userId);
    }
    
    @Override
    public int hashCode() {
        return Objects.hash(userId);
    }
    
    @Override
    public String toString() {
        return "UserData{" +
                "userId='" + userId + '\'' +
                ", userName='" + userName + '\'' +
                ", age=" + age +
                ", gender='" + gender + '\'' +
                ", email='" + email + '\'' +
                ", phone='" + phone + '\'' +
                ", city='" + city + '\'' +
                ", department='" + department + '\'' +
                ", position='" + position + '\'' +
                ", salary=" + salary +
                ", userLevel='" + userLevel + '\'' +
                ", ageGroup='" + ageGroup + '\'' +
                ", salaryLevel='" + salaryLevel + '\'' +
                ", processTime='" + processTime + '\'' +
                ", partitionDate='" + partitionDate + '\'' +
                '}';
    }
} 